package org.spamforces.rescanner;

import com.codeforces.commons.process.ThreadUtil;
import org.apache.log4j.Logger;
import org.nocturne.util.StringUtil;
import org.spamforces.dao.*;
import org.spamforces.dao.impl.*;
import org.spamforces.model.*;
import org.spamforces.utils.BroadcastHelper;
import org.spamforces.utils.SmsUtil;
import org.spamforces.utils.XMLHelper;

import javax.mail.*;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class RescannerThread implements Runnable {
    private static final Logger logger = Logger.getLogger(RescannerThread.class);

    private final AtomicBoolean running = new AtomicBoolean(true);

    @Override
    public void run() {
        try {
            EmailDao emailDao = new EmailDaoImpl();
            SmsDao smsDao = new SmsDaoImpl();

            RecipientProviderDao recipientProviderDao = new RecipientProviderDaoImpl();
            EmailPropertiesDao emailPropertiesDao = new EmailPropertiesDaoImpl();
            BroadcastDao broadcastDao = new BroadcastDaoImpl();

            while (running.get()) {
                long beforeBroadcastsPeekMills = System.currentTimeMillis();

                List<Broadcast> broadcasts = broadcastDao.peek();
                for (Broadcast broadcast : broadcasts) {
                    try {
                        SortedMap<Recipient, Map<String, String>> paramValueByKeyByRecipient = new TreeMap<>();
                        for (RecipientProvider provider : BroadcastHelper.getProviders(broadcast, recipientProviderDao)) {
                            paramValueByKeyByRecipient.putAll(XMLHelper.processXML(provider));
                        }

                        BroadcastHelper.scheduleBroadcast(
                                broadcast, paramValueByKeyByRecipient, emailDao, smsDao, true
                        );

                        if (paramValueByKeyByRecipient.isEmpty()) {
                            broadcast.setStatus(BroadcastStatus.COMPLETED);
                        } else {
                            broadcast.setStatus(BroadcastStatus.PENDING);
                        }

                        broadcastDao.update(broadcast);
                    } catch (Exception e) {
                        e.printStackTrace();
                        logger.error("Can't process broadcast.", e);
                    }
                }

                long beforeLettersPeekMills = System.currentTimeMillis();

                List<Email> emails = emailDao.peek(1000);
                List<Sms> smsList = smsDao.peek(1000);

                if (emails.isEmpty() && smsList.isEmpty()) {
                    ThreadUtil.sleep(5000L);
                    continue;
                }

                logger.info(String.format(
                        "Add %d emails and %d SMSs in %d ms.",
                        emails.size(), smsList.size(), System.currentTimeMillis() - beforeLettersPeekMills
                ));

                logger.info(String.format(
                        "Add %d broadcasts in %d ms.",
                        broadcasts.size(), beforeLettersPeekMills - beforeBroadcastsPeekMills
                ));

                long beforeSendingTimeMillis = System.currentTimeMillis();
                Set<Long> broadcastIds = new HashSet<>();

                AtomicInteger failedEmailCount = new AtomicInteger();
                sendEmails(emails, emailDao, emailPropertiesDao, broadcastIds, failedEmailCount);

                AtomicInteger failedSmsCount = new AtomicInteger();
                sendSmsList(smsList, smsDao, broadcastIds, failedSmsCount);

                for (Long broadcastId : broadcastIds) {
                    Broadcast broadcast = broadcastDao.find(broadcastId);
                    if (broadcast != null) {
                        if (!emailDao.isAnythingInQueue(broadcastId) && !smsDao.isAnythingInQueue(broadcastId)) {
                            if (broadcast.getStatus() == BroadcastStatus.PENDING) {
                                broadcast.setStatus(BroadcastStatus.COMPLETED);
                                broadcastDao.update(broadcast);
                            }
                        }
                    }
                }

                long sendingDurationMillis = System.currentTimeMillis() - beforeSendingTimeMillis;
                logger.info(String.format(
                        "Finished to send %d emails and %d SMSs in %d ms, failed email count is %d, failed SMS count is %d.",
                        emails.size(), smsList.size(), sendingDurationMillis, failedEmailCount.get(), failedSmsCount.get()
                ));
            }
        } catch (Exception e) {
            logger.error("Unexpected exception " + e.getMessage(), e);
        }
    }

    private static void sendEmails(List<Email> emails, EmailDao emailDao, EmailPropertiesDao emailPropertiesDao,
                                   Set<Long> broadcastIds, AtomicInteger failedEmailCount) {
        final Properties props = emailPropertiesDao.getDefault();
        Session mailSession;

        if (StringUtil.isEmptyOrNull(props.getProperty("mail.user"))) {
            mailSession = Session.getInstance(props);
        } else {
            mailSession = Session.getInstance(props,
                    new Authenticator() {
                        @Override
                        protected PasswordAuthentication getPasswordAuthentication() {
                            return new PasswordAuthentication(
                                    props.getProperty("mail.user"),
                                    props.getProperty("mail.password")
                            );
                        }
                    });
        }

        for (Email email : emails) {
            try {
                email.setStatus(Status.FAILED);
                emailDao.update(email);
                MimeMessage message = new MimeMessage(mailSession);

                message.setRecipient(Message.RecipientType.TO, new InternetAddress(email.getTargetEmail()));
                message.setFrom(new InternetAddress(email.getFromEmail()));
                message.setSubject(email.getSubject());
                message.setSentDate(new Date());
                message.setContent(email.getBody(), "text/html; charset=UTF-8");
                message.setHeader("Content-Type", "text/html; charset=utf-8");
                message.setHeader("Content-Transfer-Encoding", "quoted-printable");
                message.setHeader("Precedence", "bulk");

                Transport.send(message);

                /*// TODO: Do not commit, development only
                if (email.getTargetEmail().startsWith("los.ilia.ru")) {
                    Transport.send(message, message.getRecipients(Message.RecipientType.TO));
                } else {
                    ThreadUtil.sleep(10L);
                }*/

                email.setStatus(Status.DONE);
                broadcastIds.add(email.getBroadcastId());
                emailDao.update(email);

                ThreadUtil.sleep(10L);
            } catch (Exception e) {
                failedEmailCount.incrementAndGet();
                logger.warn("Can't send email to " + email.getTargetEmail() + '.', e);
                ThreadUtil.sleep(1000L);
            }
        }
    }

    private static void sendSmsList(
            List<Sms> smsList, SmsDao smsDao, Set<Long> broadcastIds, AtomicInteger failedSmsCount) {
        for (Sms sms : smsList) {
            try {
                sms.setStatus(Status.FAILED);
                smsDao.update(sms);

                SmsUtil.send(sms);

                sms.setStatus(Status.DONE);
                broadcastIds.add(sms.getBroadcastId());
                smsDao.update(sms);

                ThreadUtil.sleep(10L);
            } catch (Exception e) {
                failedSmsCount.incrementAndGet();
                logger.warn("Can't send SMS to " + sms.getPhone() + '.', e);
                ThreadUtil.sleep(1000L);
            }
        }
    }

    public final void stop() {
        running.set(false);
    }
}
